#include "cc_event_thread.h"

#define CC_EVTH_GET_MSG_TIME 50
extern cc_port ports[CC_CONFIG_CC_PORT_MAX_NUM];

static int __cc_event_thread_send(cc_event_thread_t ev_th, uint32_t target_port_id, uint32_t event, void *data, uint32_t data_size, void *user_data, cc_mn_callback cb)
{
  // 确认双端都处于连接状态
  if (ev_th == CC_NULL)
  {
    cc_log_e("ev_th is null\r\n");
    return CC_ENOMEM;
  }
  cc_port_t src_port = cc_port_get(ev_th->port_id);
  if (src_port == CC_NULL)
  {
    cc_log_e("src evth not exists\r\n", ev_th->port_id);
    return CC_ENOMEM;
  }
  if (src_port->port_id == CC_PORT_UNUSE)
  {
    cc_log_e("can't use id CC_PORT_UNUSE as a port\r\n", ev_th->port_id);
    return CC_EBUSY;
  }
  if (src_port->state != CC_PORT_STATE_CONN)
  {
    cc_log_e("src evth not connect\r\n", ev_th->port_id);
    return CC_ENOTCONN;
  }
  cc_port_t target_port = cc_port_get((e_cc_port)target_port_id);
  if (target_port_id == 0 || target_port == CC_NULL)
  {
    cc_log_e("target evth no exists now\r\n");
    return CC_ENOTCONN;
  }
  if (cc_port_get_st(target_port) != CC_PORT_STATE_CONN)
  {
    cc_log_e("target evth no connect\r\n");
    return CC_ENOTCONN;
  }
  static int retry_cnt = 0;
  cc_mnode_t node = cc_port_get_msg_node(src_port, CC_EVTH_GET_MSG_TIME);
  if (node == CC_NULL)
  {
    cc_log_e("node is null\r\n");
    return CC_ENOMEM;
  }
  if (user_data != CC_NULL)
  {
    node->user_data = user_data;
  }
  if (cb != CC_NULL)
  {
    node->callback = cb;
  }
  cc_event_msg_t event_msg = (cc_event_msg_t)node->mem;
  event_msg->event = event;
  memset(event_msg->topic, 0, sizeof(event_msg->topic));

  if (data_size < strlen((char *)event_msg->data) || data == CC_NULL)
    memset(event_msg->data, 0, strlen((char *)event_msg->data) + 1);

  if (data_size != 0)
  {
    memcpy(event_msg->data, data, data_size);
  }
  if (node == CC_NULL)
  {
    return CC_ENOMEM;
  }

  return cc_port_send(src_port, (e_cc_port)target_port_id, node);
}

static void __event_thread_entry(void *data)
{
  cc_event_thread_t evth = (cc_event_thread_t)data;
  cc_port_t port = cc_port_get(evth->port_id);
  uint8_t exe_flag = 0;
  while (1)
  {
    cc_mnode_t node = cc_port_recv(port);
    if (node != CC_NULL)
    {
      cc_mutex_lock(evth->mux);
      cc_event_msg_t event_msg = (cc_event_msg_t)node->mem;

      // topic处理
      if (event_msg->event == CC_TOPIC_EVENT)
      {
        void *temp_data;
        cc_list_foreach(&(port->sublist_head))
        {
          cc_list_get_entry(&(port->sublist_head), temp_data);
          struct cc_port_sub_list *temp = (struct cc_port_sub_list *)temp_data;
          if (strcmp(temp->topic, event_msg->topic) == 0)
          {
            uint32_t level;
            if (evth->occupy_flag)
            {
              level = cc_thread_enter_critical();
            }
            if (evth->topic_cb)
            {

              evth->topic_cb(evth, node);
            }
            if (evth->occupy_flag)
            {
              cc_thread_exit_critical(level);
            }
            exe_flag = 1;
            break;
          }
          else
          {
            exe_flag = 0;
          }
        }
      }
      else
      {
        // 事件处理
        cc_list_foreach(&evth->event_list)
        {
          void *temp_entry;
          cc_list_get_entry(&evth->event_list, temp_entry);
          cc_event_handle_t event_handle = (cc_event_handle_t)temp_entry;
          event_handle->arg = event_msg->data;
          cc_log_d("recv:%d %d\r\n", event_msg->event, event_handle->event);
          // 事件处理
          if (event_handle->event == event_msg->event)
          {
            uint32_t level;
            if (evth->occupy_flag)
            {
              level = cc_thread_enter_critical();
            }
            event_handle->fun_cb(evth, node);

            exe_flag = 1;
            if (evth->occupy_flag)
            {
              cc_thread_exit_critical(level);
            }

            break;
          }
          else
          {
            exe_flag = 0;
          }
        }
      }

      if (exe_flag != 1)
      {
        cc_log_e("port(%d) recv but not have the event(%d) attach\r\n", port->port_id, event_msg->event);
      }
      // TODO 在发送速度很快时 这边的cc_mnode_put不知道为什么要dump 用这个就不会
      cc_mempool_free(node->srchdl, node);
      cc_mutex_unlock(evth->mux);
    }
  }
}

cc_event_thread_t cc_event_thread_create(uint32_t port_id, uint8_t priority, uint32_t stack_size, uint32_t msg_num, uint32_t msg_size, cc_event_notify_cb topic_cb)
{
  cc_event_thread_t evth = (cc_event_thread_t)cc_malloc(sizeof(cc_event_thread));
  if (evth == CC_NULL)
  {
    cc_log_e("ev_th is null\r\n");
    return CC_NULL;
  }
  evth->occupy_flag = 0;
  evth->mux = cc_mutex_create();
  cc_list_init(&evth->event_list);
  if (evth->mux == CC_NULL)
  {
    cc_log_e("mux is null\r\n");
    return CC_NULL;
  }
  if (port_id != CC_PORT_UNUSE)
  {
    evth->port_id = (e_cc_port)port_id;
  }
  else
  {
    cc_log_e("can not use port CC_PORT_UNUSE to init evth\r\n");
    free(evth);
    return CC_NULL;
  }
  char temp[20];
  memset(temp, 0, sizeof(temp));
  sprintf(temp, "event:%d", port_id);
  cc_port_t evth_port = cc_port_init(evth->port_id, temp, 10, 200);
  if (evth_port == CC_NULL)
  {
    cc_log_e("evth_port is null\r\n");
    return CC_NULL;
  }
  cc_port_set_state(evth_port, CC_PORT_STATE_DISCONN);
  evth->tid = cc_thread_create_raw(temp, stack_size, priority, __event_thread_entry, evth);
  if (evth->tid == CC_NULL)
  {
    cc_log_e("tid is null\r\n");
    return CC_NULL;
  }
  evth->topic_cb = topic_cb;
  cc_port_subsribe(cc_port_get(evth->port_id), "evth_broadcast");
  cc_thread_suspend(evth->tid);
  return evth;
}

int cc_event_thread_destroy(cc_event_thread_t ev_th)
{
}



int cc_event_thread_send_data(cc_event_thread_t ev_th, uint32_t target_port_id, uint32_t event, void *data, uint32_t data_size)
{
  return __cc_event_thread_send(ev_th, target_port_id, event, data, data_size, CC_NULL, CC_NULL);
}

int cc_event_thread_send_cb(cc_event_thread_t ev_th, uint32_t target_port_id, uint32_t event, cc_mn_callback cb)
{
  return __cc_event_thread_send(ev_th, target_port_id, event, CC_NULL, 0, CC_NULL, cb);
}

int cc_event_thread_send_userdata(cc_event_thread_t ev_th, uint32_t target_port_id, uint32_t event, void *user_data)
{
  return __cc_event_thread_send(ev_th, target_port_id, event, CC_NULL, 0, user_data, CC_NULL);
}

int cc_event_thread_send_all(cc_event_thread_t ev_th, uint32_t target_port_id, uint32_t event, void *data, uint32_t data_size, void *user_data, cc_mn_callback cb)
{
  return __cc_event_thread_send(ev_th, target_port_id, event, data, data_size, user_data, cb);
}

int cc_event_thread_notify_attach(cc_event_thread_t ev_th, uint32_t event, cc_event_notify_cb cb)
{
  if (ev_th == CC_NULL)
  {
    cc_log_e("ev_th is null\r\n");
    return CC_ENOMEM;
  }
  cc_mutex_lock(ev_th->mux);
  cc_list_foreach(&ev_th->event_list)
  {
    void *temp_entry;
    cc_list_get_entry(&ev_th->event_list, temp_entry);
    cc_event_handle_t event_handle = (cc_event_handle_t)temp_entry;
    if (event == event_handle->event)
    {
      cc_log_e("repeat attach\r\n");
      cc_mutex_unlock(ev_th->mux);
      return CC_EINVAL;
    }
  }

  cc_event_handle_t evth_handle = (cc_event_handle_t)cc_malloc(sizeof(cc_event_handle));
  cc_list_node_init(&evth_handle->node);
  cc_list_node_set_entry(&evth_handle->node, evth_handle);
  evth_handle->event = event;
  evth_handle->fun_cb = cb;
  cc_list_insert_after(&ev_th->event_list, &evth_handle->node);
  cc_mutex_unlock(ev_th->mux);

  return CC_EOK;
}

int cc_event_thread_notify_detach(cc_event_thread_t ev_th, uint32_t event)
{
  if (ev_th == CC_NULL)
  {
    cc_log_e("ev_th is null\r\n");
    return CC_ENOMEM;
  }
  cc_mutex_lock(ev_th->mux);
  cc_list_foreach(&ev_th->event_list)
  {
    void *temp_entry;
    cc_list_get_entry(&ev_th->event_list, temp_entry);
    cc_event_handle_t event_handle = (cc_event_handle_t)temp_entry;
    if (event == event_handle->event)
    {
      cc_list_remove(&event_handle->node);
      free(event_handle);
      event_handle = CC_NULL;
      cc_mutex_unlock(ev_th->mux);
      return CC_EOK;
    }
  }
  cc_mutex_unlock(ev_th->mux);
  return CC_EINVAL;
}

int cc_event_thread_start(cc_event_thread_t ev_th)
{
  if (ev_th == CC_NULL)
  {
    cc_log_e("ev_th is null\r\n");
    return CC_ENOMEM;
  }
  cc_thread_resume(ev_th->tid);
  return cc_port_set_state(cc_port_get(ev_th->port_id), CC_PORT_STATE_CONN);
}

int cc_event_thread_suspend(cc_event_thread_t ev_th)
{
  if (ev_th == CC_NULL)
  {
    cc_log_e("ev_th is null\r\n");
    return CC_ENOMEM;
  }
  cc_mutex_lock(ev_th->mux);
  cc_thread_suspend(ev_th->tid);
  int status = cc_port_set_state(cc_port_get(ev_th->port_id), CC_PORT_STATE_DISCONN);
  cc_mutex_unlock(ev_th->mux);
  return status;
}

int cc_event_thread_set_occupy_status(cc_event_thread_t ev_th, uint8_t status)
{
  if (ev_th == CC_NULL)
  {
    cc_log_e("ev_th is null\r\n");
    return CC_ENOMEM;
  }
  if (!(status == 1 && status == 0))
  {
    return CC_EINVAL;
  }
  cc_mutex_lock(ev_th->mux);
  ev_th->occupy_flag = status;
  cc_mutex_unlock(ev_th->mux);
}
int cc_event_thread_get_occupy_status(cc_event_thread_t ev_th)
{
  if (ev_th == CC_NULL)
  {
    cc_log_e("ev_th is null\r\n");
    return -CC_ENOMEM;
  }
  return ev_th->occupy_flag;
}


int cc_event_thread_publish(cc_event_thread_t ev_th, const char *topic, void *data, uint32_t data_size)
{
  if (ev_th == CC_NULL)
  {
    cc_log_e("ev_th is null\r\n");
    return CC_ENOMEM;
  }

  cc_port_t src_port = cc_port_get(ev_th->port_id);
  if (src_port == CC_NULL)
  {
    cc_log_e("src port:%d not exists\r\n", ev_th->port_id);
    return CC_EBUSY;
  }
  if (src_port->port_id == CC_PORT_UNUSE)
  {
    cc_log_e("can't use id CC_PORT_UNUSE as a port\r\n", ev_th->port_id);
    return CC_EBUSY;
  }
  if (src_port->state != CC_PORT_STATE_CONN)
  {
    cc_log_e("src port:%d not connect\r\n", ev_th->port_id);
    return CC_ENOTCONN;
  }

  for (size_t i = 0; i < CC_CONFIG_CC_PORT_MAX_NUM; i++)
  {
    cc_port_t target_port = &ports[i];
    // 找到一个可以发送的端口
    if (target_port->port_id != ev_th->port_id && target_port->state == CC_PORT_STATE_CONN && target_port->port_id != CC_PORT_UNUSE)
    {
      void *temp_data;
      // 判断端口是否订阅了目标topic
      cc_list_foreach(&(target_port->sublist_head))
      {
        cc_list_get_entry(&(target_port->sublist_head), temp_data);
        struct cc_port_sub_list *temp = (struct cc_port_sub_list *)temp_data;
        // 找到目标
        if (strcmp(temp->topic, topic) == 0)
        {
          cc_mnode_t node = cc_port_get_msg_node(src_port, CC_EVTH_GET_MSG_TIME);
          // 可以考虑增加重试机制
          if (node == CC_NULL)
          {
            cc_log_e("do not have send node now\r\n");
            break;
          }
          cc_event_msg_t event_msg = (cc_event_msg_t)node->mem;
          event_msg->event = CC_TOPIC_EVENT;
          memcpy(event_msg->topic, topic, sizeof(event_msg->topic));
          if (data_size != 0)
          {
            memcpy(event_msg->data, data, data_size);
          }
          if (node == CC_NULL)
          {
            return CC_ENOMEM;
          }
          // 发送 然后break出去寻找下一个符合topic的端口
          cc_port_send(src_port, (e_cc_port)target_port->port_id, node);
          break;
        }
      }
    }
  }
  return CC_EOK;
}

int cc_event_thread_broadcast(cc_event_thread_t ev_th, void *data, uint32_t data_size)
{
  return cc_event_thread_publish(ev_th, "evth_broadcast", data, data_size);
}

int cc_event_thread_subscribe(cc_event_thread_t ev_th, const char *topic)
{
  if (ev_th == CC_NULL)
  {
    cc_log_e("ev_th is null\r\n");
    return -CC_ENOMEM;
  }
  cc_mutex_lock(ev_th->mux);
  int status = cc_port_subsribe(cc_port_get(ev_th->port_id), topic);
  cc_mutex_unlock(ev_th->mux);
  return status;
}

int cc_event_thread_unsubscribe(cc_event_thread_t ev_th, const char *topic)
{
  if (ev_th == CC_NULL)
  {
    cc_log_e("ev_th is null\r\n");
    return -CC_ENOMEM;
  }
  cc_mutex_lock(ev_th->mux);
  int status = cc_port_unsubsribe(cc_port_get(ev_th->port_id), topic);
  cc_mutex_unlock(ev_th->mux);
  return status;
}